package com.showdor.springboot.kafka;

/**
 * @author showdor
 * @email chuanqi@outlook.com
 * @date 2023/7/5
 */

import com.showdor.springboot.ConsumerRecordDAO;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.support.Acknowledgment;

import java.util.Optional;

/**
 * 多线程kafka消费类
 */
public class OneWork implements Runnable {
    //日志类
    private static final Logger logger = LoggerFactory.getLogger(OneWork.class);

    private ConsumerRecord<String, String> consumerRecord;
    private ConsumerRecordDAO dao;
    private Acknowledgment ack;

    public OneWork(ConsumerRecordDAO dao, ConsumerRecord record, Acknowledgment ack) {
        this.consumerRecord = record;
        this.dao = dao;
        this.ack = ack;
    }

    @Override
    public void run() {
        try {
            logger.info("获取数据：" + this.consumerRecord.value());
            Optional<?> optional = Optional.ofNullable(consumerRecord.value());
            if (optional.isPresent()) {
                Object msg = optional.get();
                dao.create(consumerRecord.key(), msg, ack);
            }
            //执行自己的逻辑
        } catch (Exception e) {
            logger.info("异常错误信息：" + e.getMessage());
        }
    }
}
